package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo26PageRank {
  def main(args: Array[String]): Unit = {

    // 加载数据
    val sc: SparkContext = new SparkContext(new SparkConf().setAppName("Demo26PageRank").setMaster("local"))

    val sourcePageRDD: RDD[String] = sc.textFile("Spark/data/page_rank.txt")
    // 统计页面的数量
    val pages: Long = sourcePageRDD.count()


    // 预处理数据 将每一行数据按照 -> 进行分隔
    val prePageRDD: RDD[(String, List[String])] = sourcePageRDD.map(line => {
      val splits: Array[String] = line.split("->")
      val page: String = splits(0)
      val desPages: List[String] = splits(1).split(",").toList
      (page, desPages)
    })

    // 给每个页面附上初始的PR值 1
    var oldPagePrRDD: RDD[(String, (List[String], Double))] = prePageRDD.map(t2 => (t2._1, (t2._2, 1.0)))
    oldPagePrRDD.foreach(println)

    var flag: Boolean = true
    // 引入阻尼系数
    val q: Double = 0.85
    // 统计迭代次数
    var i: Int = 1
    while (flag) {
      // 将Pr值平均分给每一个指向的页面
      val newPagePrRDD: RDD[(String, (List[String], Double))] = oldPagePrRDD.flatMap {
        case (page: String, (desPages: List[String], pr: Double)) =>
          desPages.map(desPage => (desPage, pr / desPages.size))
      }
        // 计算每个页面新的Pr值
        .reduceByKey(_ + _)
        // 关联 将新的Pr值赋给每个页面
        .join(oldPagePrRDD)
        .map {
          case (page: String, (newPr: Double, (desPages: List[String], oldPr: Double))) =>
            (page, (desPages, ((1 - q) / pages) + q * newPr))
        }


      // 计算误差
      val avgDiff: Double = newPagePrRDD
        .join(oldPagePrRDD)
        .map {
          case (page: String, ((desPage1: List[String], newPr: Double), (desPage2: List[String], oldPr: Double))) =>
            // 计算当前的Pr值和上一次Pr值的误差
            Math.abs(newPr - oldPr)
        }.sum() / pages // 计算平均误差
      println(s"第${i}次迭代的结果")
      newPagePrRDD.foreach(println)
      println(s"平均误差为：${avgDiff.toString}")

      // 通过误差的判断决定是否收敛 --> 是否停止循环
      if (avgDiff < 0.0001) {
        flag = false
      }
      // 开始新的一次迭代
      oldPagePrRDD = newPagePrRDD

      i += 1

    }


  }

}
